/*******************************************************************************
 * Copyright (c) 2010 Mikhail Garber.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the GNU Lesser Public License (LGPL)
 * which accompanies this distribution
 * 
 * Contributors:
 *     Mikhail Garber - initial API and implementation
 ******************************************************************************/
package plumber.coordinator;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.UUID;


import plumber.common.WorkItem;
import plumber.common.WorkItem.Status;

/**
 * example of Coordinator that runs in-process (same VM as both Clients submitting work
 * and a WorkManager executing work using pool of workers). Suitable for small, non-distributed systems
 * 
 * @author mgarber
 *
 */
public class InProcessCoordinator extends AbstractCoordinator implements
		CoordinatorClient, CoordinatorServer {

	/**
	 * local store of WorkItems
	 */
	private List<WorkItem> allItems = new ArrayList<WorkItem>();
	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorServer#getCandidates(java.lang.String, int)
	 */
	@Override
	public List<String> getCandidates(String namespace, int upto) {
		List<WorkItem> result = new ArrayList<WorkItem>();
		synchronized (allItems) {
			int i=0;
			for (WorkItem item : allItems) {
				if (item.getNamespace().equals(namespace) && 
						Status.AVAILABLE == item.getStatus() &&
						item.getScheduled().before(new Date())) {
					result.add(item);
					i++;
					if(i>upto) {
						break;
					}
				}
			}
		}
		Collections.sort(result);
		return this.contextConverter.toStorableStrings(result);
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorServer#claimCandidate(java.lang.String)
	 */
	@Override
	public boolean claimCandidate(String id) {
		synchronized (allItems) {
			for (WorkItem item : allItems) {
				if (item.getId().equals(id)
						&& Status.AVAILABLE == item.getStatus()) {
					item.setStatus(Status.IN_PROGRESS);
					logger.info("claimed:" + id);
					return true;
				}
			}
		}
		logger.info("could not claim:" + id);
		return false;
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorClient#submitWork(java.lang.String, java.lang.String, java.lang.String)
	 */
	@Override
	public String submitWork(String  namespace, String context, String flowName) {
		return submitWork(namespace, context, flowName, new Date());
	}
	
	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorClient#submitWork(java.lang.String, java.lang.String, java.lang.String, java.util.Date)
	 */
	@Override
	public String submitWork(String namespace, String context, String flowName, Date scheduled) {
		
		WorkItem item = makeNewItem(context, flowName, null, scheduled, UUID
				.randomUUID().toString(), namespace);
		synchronized (allItems) {
			allItems.add(item);
		}
		logger.info("submitted to " + flowName);
		return item.getRequestId();
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorClient#getHistory(java.lang.String)
	 */
	@Override
	public List<String> getHistory(String requestId) {
		List<WorkItem> result = getItemHistory(requestId);
		return this.contextConverter.toStorableStrings(result);
	}

	private List<WorkItem> getItemHistory(String requestId) {
		List<WorkItem> result = new ArrayList<WorkItem>();
		synchronized (allItems) {
			for (WorkItem item : allItems) {
				if (item.getRequestId().equals(requestId)) {
					result.add(item);
				}
			}
		}
		Collections.sort(result);
		return result;
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorClient#checkComplete(java.lang.String)
	 */
	@Override
	public boolean checkComplete(String requestId) {
		List<String> items = new ArrayList<String>(getHistory(requestId));
		return hasStatus(items, Status.FINAL);
	}

	private boolean hasStatus(List<String> items, Status status) {
		return (findByStatus(this.contextConverter.itemsFromStorableString(items), status) != null);
	}

	private WorkItem findByStatus(List<WorkItem> items, Status status) {
		for (WorkItem item : items) {
			if (status == item.getStatus()) {
				return item;
			}
		}
		return null;
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorClient#checkFailed(java.lang.String)
	 */
	@Override
	public boolean checkFailed(String requestId) {
		return hasStatus(getHistory(requestId), Status.FAILED);
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorClient#getResult(java.lang.String)
	 */
	@Override
	public String getResult(String requestId) {
		WorkItem item = findByStatus(this.contextConverter.itemsFromStorableString(getHistory(requestId)), Status.FINAL);
		return (item == null) ? null : item.getContext();
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorServer#advanceWork(java.lang.String, java.lang.String, java.lang.String, plumber.common.WorkItem.Status, java.util.Date)
	 */
	@Override
	public void advanceWork(String context, String oldItemId,
			String nextStepId, Status oldStatus, Date scheduled) {
		logger.info("advancing work: " + context + " "
				+ oldStatus + " " + nextStepId);
		synchronized (allItems) {
			for (WorkItem item : allItems) {
				if (item.getId().equals(oldItemId)) {
					item.setStatus(oldStatus);
					item.setContext(context);
					logger.info("changed status to " + oldStatus);
					if (nextStepId != null) {
						WorkItem newItem = makeNewItem(context, item.getFlowName(), nextStepId,
								scheduled, item.getRequestId(), item.getNamespace());
						allItems.add(newItem);
						logger.info("created new item");
					}
					logger.info("all:" + allItems);
					return;
				}
			}
			throw new IllegalStateException("could not find item by id");
		}
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorClient#getAll(java.lang.String, int)
	 */
	@Override
	public List<String> getAll(String namespace, int upto) {
		List<WorkItem> result = new ArrayList<WorkItem>();
		synchronized (allItems) {
			result.addAll(allItems);
		}
		Collections.sort(result);
		List<WorkItem> realResult = new ArrayList<WorkItem>();
		int added = 0;
		for(WorkItem item : result) {
			if(added >= upto) {
				break;
			}
			if(item.getNamespace().equals(namespace)) {
				realResult.add(item);
				added++;
			}
		}
		return this.contextConverter.toStorableStrings(realResult);
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorClient#suspend(java.lang.String)
	 */
	@Override
	public boolean suspend(String requestId) {
		return changeStatusOfLastItem(requestId, Status.AVAILABLE, Status.SUSPENDED);
		
	}

	private boolean changeStatusOfLastItem(String requestId, Status from, Status to) {
		List<WorkItem> all = this.getItemHistory(requestId);
		WorkItem last = (all.size() > 0)? all.get(0) : null;
		if(last == null) {
			return false;
		}
		
		synchronized (allItems) {
			if( last.getStatus() != from) {
				return false;
			}
			last.setStatus(to);
		}
		return true;
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorClient#resume(java.lang.String)
	 */
	@Override
	public boolean resume(String requestId) {
		return changeStatusOfLastItem(requestId, Status.SUSPENDED, Status.AVAILABLE);
	}
}
